package com.imooc.ecommerce.rocket;

import com.alibaba.fastjson.JSON;
import com.imooc.ecommerce.vo.QinyiMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

/**
 * @description: 大部分情况都是同步
 * @author: sym
 * @create: 2021-11-29 23:44
 **/
@Slf4j
@Component
public class RocketMqProducer {
    /** 类似 Kafka 中的 topic, 默认的读写队列都是4个 */
    private static final String TOPIC = "imooc-study-rocketmq";

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void sendMessageSync(String value){
        SendResult sendResult = rocketMQTemplate.syncSend(TOPIC,value);
        log.info("sendMessageWithValue result:{}", JSON.toJSONString(sendResult));
        SendResult syncSendOrderly = rocketMQTemplate.syncSendOrderly(
                TOPIC, value, "sym"
        );
        log.info("sendMessageWithValue orderly result: [{}]",
                JSON.toJSONString(syncSendOrderly));
    }

    public void sendMessageAsync(String key,String value){

        Message<String> message= MessageBuilder.withPayload(value)
                // 带有key的一条消息，key是索引,key是多个用空格分开，kafka的key是带有分区效果
                .setHeader(RocketMQHeaders.KEYS,key).build();

        // 异步发送消息，一般需要设定回调
        rocketMQTemplate.asyncSend(TOPIC, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("sendMessageAsync success result: [{}]",
                        JSON.toJSONString(sendResult));
            }

            @Override
            public void onException(Throwable e) {
                log.error("sendMessageAsync failure: [{}]", e.getMessage(), e);
            }
        });
    }

    // 带有tag,发送的java对象
    public void sendMessageSyncWithTag(String value,String tag){
        QinyiMessage qinyiMessage = JSON.parseObject(value, QinyiMessage.class);
        SendResult sendResult = rocketMQTemplate.syncSend(String.format("%s:%s", TOPIC, tag), qinyiMessage);
        log.info("sendMessageSyncWithTag result: [{}]", JSON.toJSONString(sendResult));
    }

    /**
     * <h2>使用同步的方式发送消息, 带有 key 和 tag</h2>
     * */
    public void sendMessageWithAll(String key, String tag, String value) {

        Message<String> message = MessageBuilder.withPayload(value)
                .setHeader(RocketMQHeaders.KEYS, key).build();
        SendResult sendResult = rocketMQTemplate.syncSend(
                String.format("%s:%s", TOPIC, tag),
                message
        );
        log.info("sendMessageWithAll result: [{}]", JSON.toJSONString(sendResult));
    }
}
